package com.bonc.wkafka.utils.producer;

import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.bonc.wkafka.utils.KafkaProps.SendPattern;
import com.bonc.wkafka.utils.producer.handler.KafkaExceptionHandler;
import com.bonc.wkafka.utils.producer.handler.KafkaSucMetadataHandler;

/**
 * kafka producer</br>
 * 封装了以下几个功能：</br>
 * 1. 封装了producer的配置，给出几个生产模式的一键配置；</br>
 * 2. 给出三种消息发送模型：发送并遗忘、同步发送和异步发送；</br>
 * 3. 给出发送失败消息的处理模型：同步处理，线程池处理，线程池+缓存处理，信号量线程池处理；</br>
 * @author WYB
 *
 */
public class WKafkaProducer<K, V> {
	
	private static final Logger logger = LoggerFactory.getLogger(WKafkaProducer.class);
	
	private final KafkaProducer<K, V> producer;
	
	private KafkaSucMetadataHandler sucMetaHandler;
	
	private KafkaExceptionHandler<K, V> exceptionHandler;

	private KafkaProducerProps props;

	private SendPattern sendPattern;
	
	public WKafkaProducer(KafkaProducerProps props, KafkaSucMetadataHandler sucMetaHandler, KafkaExceptionHandler<K, V> exceptionHandler) {
		this.props = props;
		producer =  new KafkaProducer<K, V>(props.getProps(), props.getKeySerializer(), props.getValueSerializer());
		this.sucMetaHandler = sucMetaHandler;
		this.exceptionHandler = exceptionHandler;
		this.sendPattern = props.getSendPattern();
	}
	
	
	/**
	 * 三种方式的生产消息
	 * @param topic
	 * @param key
	 * @param value
	 */
	public void produce(String topic, K key, V value) {
		ProducerRecord<K, V> record = new ProducerRecord<K, V>(topic, key, value);
		if (sendPattern==SendPattern.FORGOTTEN) {
			producer.send(record);
		}else if (sendPattern==SendPattern.SYNC) {
			try {
				// 阻塞等待broker的响应
				// 如果broker返回错误，get()方法会抛出异常
				RecordMetadata metadata = (RecordMetadata) producer.send(record).get();
				logger.info("topic: %s, partition: %s, offset: %d, serializedKeySize: %s, serializedValueSize: %s\n", 
						metadata.topic(), metadata.partition(), metadata.offset(), metadata.serializedKeySize(), metadata.serializedValueSize());
				if (sucMetaHandler!=null)
					sucMetaHandler.handle(metadata);
			} catch (InterruptedException | ExecutionException e) {
				exceptionHandler.handle(e, record);
			}
		}else if (sendPattern==SendPattern.ASYNC) {
			if (exceptionHandler.needAcquire()) {
				exceptionHandler.acquire();
			}
			// Callback是运行在Sender线程中（每条消息都会触发一次），务必要快速返回，以免影响消息发送的速度
			producer.send(record, new Callback() {
				
				@Override
				// 服务端处理record时产生的exception
				// 所有生产的消息是单线程循环调用onCompletion的，如果上一次onCompletion阻塞，会影响下一次调用
				public void onCompletion(RecordMetadata metadata, Exception exception) {
					exceptionHandler.handle(exception, record);
				}
			});
		}
	}

	/**
	 * send nsync（batch send）
	 * 测试环境使用，Sender线程容易阻塞，性能不高（因为对成功的消息也做了处理）
	 * @param topic
	 * @param key
	 * @param value
	 * @param sucMetaHandler
	 * @param exceptionHandler
	 * @throws ClassNotFoundException
	 */
	private void produceAsyncTest(String topic, K key, V value) {
		ProducerRecord<K, V> record = new ProducerRecord<>(topic, key, value);
		// Callback是运行在Sender线程中（每条消息都会触发一次），务必要快速返回，以免影响消息发送的速度
		producer.send(record, new Callback() {
			
			@Override
			// 服务端处理record时产生的exception
			public void onCompletion(RecordMetadata metadata, Exception exception) {
				if (exception != null) {
					exceptionHandler.handle(exception, record);
				} else {
					sucMetaHandler.handle(metadata);
				}
			}
		});
	}
	
	public void close() {
		if (producer!=null) {
			producer.close();
		}
		
	}
}
